package com.peng.sparktest.sparkcore

import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

object SparkApiTest07_Other {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("other_test")
    val context = new SparkContext(conf)
    context.setLogLevel("ERROR")

    /**
     * 广播变量
     */
    //    val list: RDD[String] = context.parallelize(Array("hello hi", "hi hello", "world me"))
    //    val result: Array[String] = list.flatMap(_.split(" ")).
    //      map((_, 1)).reduceByKey(_ + _).sortBy(_._2, false).keys.take(2)
    //使用的这个result，属于闭包进去的变量，未来会执行多少次，这个变量就会跟随着被序列化反序列化传输多少次
    //    val res: Array[(String, Int)] = result.filter(item => result.contains(item))
    //为了解决这种数据频繁使用的问题，可以使用broadcast，进行广播变量，将变量交由blockManager存储管理
    //这样，未来某一台executor需要这个数据的时候，如果发现本地没有，则可以向driver进行拉取数据，并且只需要进行一次拉取，之后的使用，都是基于本地的
    //另外，不论是广播变量，还是单纯这种闭包，都是map的一次垂直join 而不用发生shuffle
    //    val values: Broadcast[Array[String]] = context.broadcast(result)
    //    result.filter(item => values.value.contains(item)).foreach(println)


    /**
     * 累加器
     */

    val data: RDD[Int] = context.parallelize(1 to 10, 2)
    var i = 0
    //类似于linux父线程fork子线程，只会把父线程当前的值复制一份给子线程，子线程的改变不会对父线程造成影响
    //所以，不同task都基于一个开始的值进行计算，完毕后，也不会对值进行召回
    //但有时我们需要一些计数的操作，因此 可以使用longAccumulator这种 做统计使用
    val counter: LongAccumulator = context.longAccumulator("counter")
    data.foreach(x => {
      i += 1
      counter.add(1)
    })
    println(i)
    println(counter.sum)
    println(counter.avg)
    println(counter.count)


    while (true) {

    }
  }

}
